Nettoyage et Traitement des Données NASA

Introduction

Ce script Python permet de charger, nettoyer et traiter des données de lactation et météorologiques à l'aide de l'API NASA POWER.

Étapes du script

1. Bibliothèques Requises

Ce script nécessite les bibliothèques suivantes :


import time  
import os  
import json  
import requests 
import pandas as pd  
import csv  
from datetime import timedelta

time : pour gérer les délais et les temporisations.
os : pour interagir avec le système de fichiers (création de dossiers, suppression de fichiers).
json : pour manipuler les données JSON.
requests : pour effectuer des requêtes HTTP vers l'API de la NASA.
pandas : pour manipuler et analyser les données tabulaires (DataFrames).
csv : pour lire et écrire des fichiers CSV.
datetime : pour manipuler les dates.

2. Chargement des Données

upload_df charge les données des fichiers fermes.txt et lactation_data.txt dans des DataFrames pandas. Elle renomme la colonne ferme en farm dans le DataFrame des fermes et définit les noms de colonnes pour le DataFrame des lactations.


xxxxxxxxxx
def upload_df(file_farm, file_lactating):
    columns = [
        "id_ewe", "farm_year_month", "age_GB_nb_GB", "nb_alive_lamb",
        "intervale_GB-1st_milk_recording", "date_of_GB", "milk_production_after_120_days", "farm"
    ]
    
    fermes_df = pd.read_csv(file_farm, sep='\s+').rename(columns={'ferme': 'farm'})
    lactations_df = pd.read_csv(file_lactating, sep='\s+', names=columns)
    
    return fermes_df, lactations_df

3. Fusion des Données

Cette fonction fusionne les DataFrames des fermes et des lactations en utilisant la colonne farm comme index.


xxxxxxxxxx
def join_dataframe(fermes_df, lactations_df, index):
    return lactations_df.set_index(index).join(fermes_df.set_index(index))

4. Calcul des Intervalles de Lactation

lactation_interval calcule la date de fin de lactation (date de mise bas +120 jours) pour chaque enregistrement et convertit les dates au format YYYYMMDD.


xxxxxxxxxx
def lactation_interval(df, date_col, new_col):
    df[date_col] = df[date_col].apply(lambda x: str(x).zfill(8))
    df[date_col] = pd.to_datetime(df[date_col], format="%d%m%Y", errors="coerce")
    
    df[new_col] = df[date_col] + timedelta(days=120)
    
    df[date_col] = df[date_col].dt.strftime('%Y%m%d')
    df[new_col] = df[new_col].dt.strftime('%Y%m%d')
    
    return df

5. Création du Répertoire de Sortie

Cette fonction crée un répertoire pour stocker les fichiers de sortie JSON si celui-ci n'existe pas.


xxxxxxxxxx
def create_output_directory(directory):
    os.makedirs(directory, exist_ok=True)
    return directory

6. Récupération des Données Météorologiques NASA

Cette fonction interroge l'API NASA POWER pour récupérer des données météorologiques basées sur des coordonnées géographiques et une période donnée.


x
def fetch_nasa_data(latitude, longitude, start, end, retries=6):
    url = (f"https://power.larc.nasa.gov/api/temporal/daily/point?parameters=T2M,T2MDEW,T2MWET,TS,RH2M,PRECTOTCORR,WS2M,WD2M"
           f"&community=RE&longitude={longitude}&latitude={latitude}&start={start}&end={end}&format=JSON")

    for attempt in range(retries):
        try:
            response = requests.get(url, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Erreur lors de la récupération des données NASA : {e}")
            if attempt < retries - 1:
                print("Nouvelle tentative dans 5 secondes...")
                time.sleep(5)
            else:
                print("Échec après 6 tentatives. Passage à la donnée suivante.")
                return None

7. Traitement des Données NASA

Cette fonction traite les données météorologiques en associant chaque exploitation à ses données climatiques sur la période de lactation. Elle créé le tableau csv avec les données météorologiques correspondantes à chaque lactation.


xxxxxxxxxx
def process_nasa_data(table_joint, output_dir):
    columns = [
        "id_ewe", "farm_year_month", "age_GB_nb_GB", "nb_alive_lamb",
        "intervale_GB-1st_milk_recording", "date_of_GB",
        "milk_production_after_120_days", "farm",
        "T2M", "T2MDEW", "T2MWET", "TS",
        "RH2M", "PRECTOTCORR", "WS2M", "WD2M"
    ]
    
    with open("donnees_moyennes.csv", 'a', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=columns)
        if f.tell() == 0:
            writer.writeheader()

        grouped_data = table_joint.groupby('farm')
        
        for farm_id, group in grouped_data:
            latitude = group['lat'].iloc[0]
            longitude = group['long'].iloc[0]
            start_farm = group['date_of_GB'].min()
            end_farm = group['date_120'].max()
            
            content = fetch_nasa_data(latitude, longitude, start_farm, end_farm)
            
            if not content:
                writer.writerow({"farm": farm_id})
                continue
            
            filename = f"data_{latitude}_{longitude}_{farm_id}.json"
            filepath = os.path.join(output_dir, filename)
            with open(filepath, 'w') as json_file:
                json.dump(content, json_file, indent=4)
            
            os.remove(filepath)
            time.sleep(2) #Attente de 2 seconde avant de relancer

8. Exécution du Script

L'exécution du script commence par le chargement des données, la fusion des DataFrames, le calcul des intervalles de lactation et enfin le traitement des données météorologiques.


xxxxxxxxxx
if __name__ == "__main__":
    fermes_df, lactations_df = upload_df("fermes.txt", "lactation_data.txt")
    table_joint = join_dataframe(fermes_df, lactations_df, 'farm')
        table_joint = lactation_interval(
        table_joint, "date_of_GB", "date_120")
    output_dir = create_output_directory("donnees_meteo")
    process_nasa_data(table_joint, output_dir)

Conclusion

Ce script permet une gestion automatisée des données de lactation et météorologiques en combinant diverses sources de données et en les nettoyant efficacement.